package com.sui.bigdata.flink.sql.function.aggregate;

import com.sui.bigdata.flink.sql.function.util.PhoneHHIAccum;

import org.apache.flink.table.functions.AggregateFunction;

import java.math.BigDecimal;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * @author LiuHuan
 * @date 2019/9/5 20:06
 * @description
 * @reviewer
 */
public class UserknPhoneIspRate extends AggregateFunction<Double, PhoneHHIAccum> {
    @Override
    public PhoneHHIAccum createAccumulator() {
        return new PhoneHHIAccum();
    }

    @Override
    public Double getValue(PhoneHHIAccum phoneHHIAccum) {
        Double result = 0.0000;
        Integer count = 0;
        if (!(phoneHHIAccum == null || phoneHHIAccum.getCount() == 0 || phoneHHIAccum.getGroups().isEmpty())) {
            Set<Map.Entry<String, Integer>> entries = phoneHHIAccum.getGroups().entrySet();
            for (Map.Entry<String, Integer> entry : entries) {
                count += entry.getValue();
            }
        }
        if (phoneHHIAccum.getCount() > 0) {
            Double newResult = 1 - (double) count / phoneHHIAccum.getCount();
            BigDecimal bigDecimal = new BigDecimal(newResult);
            if (newResult != null && newResult >= 0) {
                result = bigDecimal.setScale(4, BigDecimal.ROUND_HALF_UP).doubleValue();
            }
        }
        return result;
    }

    public void accumulate(PhoneHHIAccum phoneHHIAccum, String id) {
//        String[] ISPNum = {"133" ,"149" ,"153" ,"173" ,"177" ,"180" ,"181" ,"189" ,"191" ,"199" ,"130", "131" ,"132" ,"145" ,"155" ,"156" ,"166" ,"171" ,"175" ,"176" ,"185" ,"186", "134" ,"135", "136" ,"137" ,
//                "138" ,"139", "147", "150" ,"151" ,"152" ,"157" ,"158" ,"159" ,"172" ,"178", "182" ,"183", "184" ,"187" ,"188", "198" ,"165", "170" ,"167","162"};
        phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
        if (id != null) {
            if (phoneHHIAccum.getGroups().get(id) != null) {
                phoneHHIAccum.getGroups().put(id, phoneHHIAccum.getGroups().get(id) + 1);
            } else {
                phoneHHIAccum.getGroups().put(id, 1);
            }
        }
    }


    public void merge(PhoneHHIAccum phoneHHIAccum, Iterable<PhoneHHIAccum> it) {
        Iterator<PhoneHHIAccum> iter = it.iterator();
        while (iter.hasNext()) {
            PhoneHHIAccum a = iter.next();
            phoneHHIAccum.setGroups(a.getGroups());
            phoneHHIAccum.setCount(a.getCount());
            phoneHHIAccum.setRepetition(a.getRepetition());
        }
    }
}
